package com.demo.official.batch;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;

public class BatchConsumer {

    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("batch_consumer_group");
        consumer.setNamesrvAddr("127.0.0.1:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe("batch_topic", "*");

        /*
        Consumer的pullBatchSize属性与consumeMessageBatchMaxSize属性是否设置的越大越好？当然不
        是。
        pullBatchSize值设置的越大，Consumer每拉取一次需要的时间就会越长，且在网络上传输出现
        问题的可能性就越高。若在拉取过程中若出现了问题，那么本批次所有消息都需要全部重新拉
        取。
        consumeMessageBatchMaxSize值设置的越大，Consumer的消息并发消费能力越低，且这批被消
        费的消息具有相同的消费结果。因为consumeMessageBatchMaxSize指定的一批消息只会使用一
        个线程进行处理，且在处理过程中只要有一个消息处理异常，则这批消息需要全部重新再次消费
        处理
         */
        // 指定每次可以消费10条消息，默认为1
        consumer.setConsumeMessageBatchMaxSize(10);
        // 指定每次可以从Broker拉取40条消息，默认为32
        consumer.setPullBatchSize(40);

        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg : msgs) {
                //bytes转String才能明显的看到输出 https://stackoverflow.com/questions/8512121/utf-8-byte-to-string
                System.out.println(new String(msg.getBody(), StandardCharsets.UTF_8));
            }
            // 消费成功的返回结果
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            // 消费异常时的返回结果
            // return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        });

        consumer.start();
        System.out.println("Consumer Started");
    }
}
